package com.atguigu.champter8;

import com.atguigu.beans.AdsClickLog;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.text.SimpleDateFormat;
import java.time.Duration;

//同一用户当天对同一广告点击超过100词,加入黑名单报警,并不在统计
public class Flink05_Project_High_Ads {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 9999);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        SingleOutputStreamOperator<String> normal = env
                .readTextFile("datas/input/AdClickLog.csv")
                .map(line -> {
                    String[] data = line.split(",");
                    return new AdsClickLog(Long.parseLong(data[0]), Long.parseLong(data[1]), data[2], data[3], Long.parseLong(data[4]) * 1000);
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<AdsClickLog>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner((clc, ts) -> clc.getTimestamp())
                )
                .keyBy(new KeySelector<AdsClickLog, String>() {
                    @Override
                    public String getKey(AdsClickLog value) throws Exception {
                        return value.getUserId() + "-----" + value.getAdsId();
                    }
                })
                .process(new KeyedProcessFunction<String, AdsClickLog, String>() {


                    private ValueState<String> yesterday;
                    private ValueState<Boolean> flag;
                    private ReducingState<Long> adsCli;

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        adsCli = getRuntimeContext().getReducingState(new ReducingStateDescriptor<Long>("adsCli", new ReduceFunction<Long>() {
                            @Override
                            public Long reduce(Long value1, Long value2) throws Exception {
                                return value1 + value2;
                            }
                        }, Long.class));

                        flag = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("flag", Boolean.class));

                        yesterday = getRuntimeContext().getState(new ValueStateDescriptor<String>("yesterday", String.class));
                    }

                    @Override
                    public void processElement(AdsClickLog value,
                                               Context ctx,
                                               Collector<String> out) throws Exception {

                        String today = new SimpleDateFormat("yyyyMMdd").format(value.getTimestamp());

                        if (!today.equals(yesterday.value())){
                            System.out.println(today + " 封印解除");
                            adsCli.clear();
                            flag.clear();
                            yesterday.update(today);
                        }


                        if (flag.value() == null){

                            adsCli.add(1L);
                        }

                        if (adsCli.get() < 100) {
                            out.collect(ctx.getCurrentKey() + "  点击次数为  " + adsCli.get());
                        } else {
                            if (flag == null){
                                ctx.output(new OutputTag<String>("blackList") {},
                                        "警告 :" + ctx.getCurrentKey() + "  点击数超过99,行为异常加入黑名单");
                                flag.update(true);
                            }

                        }




                    }
                });

        normal.print("normal");
        normal.getSideOutput(new OutputTag<String>("blackList"){}).print("blackList");


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
